package com.zs.toolbox.web2j.system.message.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.zs.toolbox.web2j.system.message.controller.WtMessageController;
import com.zs.toolbox.web2j.system.message.domain.request.WtSendEmailRequest;
import com.zs.toolbox.web2j.system.message.domain.request.WxSendSmsVcodeRequest;
import com.zs.toolbox.web2j.system.message.mq.WtMessageTagConst;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * @author shuai.zhang  on 2018/9/26
 * @description
 * @deprecated 已被 rocketmq start替代
 */

@Slf4j
public class WtRocketMQMessageConsumerImpl {
    @Value("${message.queue.rocketmq.nameserver}")
    String nameServer;

    @Value("${message.queue.rocketmq.group}")
    String consumerGroup;

    @Value("${message.queue.rocketmq.topic}")
    String consumerTopic;

    DefaultMQPushConsumer consumer;

    @Autowired
    WtMessageController wtMessageController;

    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setNamesrvAddr(nameServer);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                log.debug("recv sms/emal send message {}", msg);
                try {
                    switch (msg.getTags()) {
                        case WtMessageTagConst.MESSSAGE_TAG_EMAIL:
                            WtSendEmailRequest request = JSON.parseObject(msg.getBody(), WtSendEmailRequest.class);
                            request.setIsSync(true);
                            wtMessageController.sendEmail(request);
                            break;
                        case WtMessageTagConst.MESSSAGE_TAG_SMS_LOGIN:
                            wtMessageController.sendPhoneLoginVCodeSms(JSON.parseObject(msg.getBody(), WxSendSmsVcodeRequest.class));
                            break;
                        case WtMessageTagConst.MESSSAGE_TAG_SMS_REGISTER:
                            wtMessageController.sendPhoneRegisterVCodeSms(JSON.parseObject(msg.getBody(), WxSendSmsVcodeRequest.class));
                            break;
                        case WtMessageTagConst.MESSSAGE_TAG_SMS_FINDPASSWORD:
                            wtMessageController.sendPhoneFindPasswordVCodeSms(JSON.parseObject(msg.getBody(), WxSendSmsVcodeRequest.class));
                            break;
                    }
                } catch (Exception e) {
                    log.warn(String.format("conumer message{} error", msg), e);
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        startConsume();
        log.info("message consumer rocketmq nameserver<{}>,group<{}> start finish [ok]", nameServer, consumerGroup);

    }

    @PreDestroy
    public void deinit() {
        consumer.shutdown();
        log.info("message consumer rocketmq nameserver<{}>,group<{}> shutdown finish [ok]", nameServer, consumerTopic);
    }

    public void startConsume() {
        try {
            consumer.subscribe(consumerTopic, "*");
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
            throw new RuntimeException("start rocketmq consumer error");
        }
    }
}
